下文|主要指_SparkMLlib算法调用展示平台及其实现过程

篇首语：本文由编程笔记#小编为大家整理，主要介绍了Spark MLlib算法调用展示平台及其实现过程相关的知识，希望对你有一定的参考价值。

1. 软件版本&＃xff1a;

IDE&＃xff1a;Intellij IDEA 14&＃xff0c;
Java&＃xff1a;1.7&＃xff0c;Scala&＃xff1a;2.10.6&＃xff1b;Tomcat&＃xff1a;7&＃xff0c;CDH&＃xff1a;5.8.0&＃xff1b; 
Spark&＃xff1a;1.6.0-cdh5.8.0-hadoop2.6.0-cdh5.8.0 &＃xff1b; 
Hadoop&＃xff1a;hadoop2.6.0-cdh5.8.0&＃xff1b;(使用的是CDH提供的虚拟机)

2. 工程下载及部署&＃xff1a;

Scala封装Spark算法工程&＃xff1a;https://github.com/fansy1990/Spark_MLlib_Algorithm_1.6.0.git &＃xff1b;
调用Spark算法工程&＃xff1a;https://github.com/fansy1990/Spark_MLlib_1.6.0_.git &＃xff1b;
部署&＃xff08;主要针对Spark_MLlib_1.6.0工程&＃xff09;&＃xff1a;

1&＃xff09;配置好db.properties中相应用户名密码&＃xff0f;数据库等参数&＃xff1b;
2&＃xff09;第一次启动tomcat&＃xff0c;修改hibernate.cfg.xml文件中的hibernate.hbm2ddl.auto值为create&＃xff0c;第二次启动修改为update&＃xff1b;
3) 打开集群参数页面&＃xff0c;点击初始化&＃xff0c;初始化集群参数&＃xff0c;如果集群参数和当前集群不匹配&＃xff0c;那么需要做相应修改&＃xff1b;
暂时考虑使用配置文件的方式来配置集群参数&＃xff0c;如果要调整为数据库配置&＃xff0c;那么修改Utisl.dbOrFile参数即可&＃xff1b;即&＃xff0c;暂时只需修改utisl.properties文件&＃xff1b;
4&＃xff09;拷贝Spark_MLlib_Algorithm_1.6.0工程生成的算法到到3&＃xff09;中spark.jar所在路径&＃xff1b;
5&＃xff09;拷贝集群中的yarn-site.xml到3&＃xff09;中spark.files所在路径&＃xff1b;
6&＃xff09;拷贝spark-assembly-1.6.0-cdh5.8.0-hadoop2.6.0-cdh5.8.0.jar到3&＃xff09;中spark.yarn.jar所在路径&＃xff1b;

3. 工程实现原理&＃xff1a;

3.1 Scala封装Spark算法工程&＃xff1a;

3.1.1 工程目录

1. 工程目录如下所示&＃xff1a;



其中&＃xff0c;data目录为所有的测试数据所在目录&＃xff0c;这里针对不同的算法建立了不同的目录&＃xff0c;主要有5类&＃xff1a;分类与回归&＃xff0f;聚类&＃xff0f;协同过滤&＃xff0f;降维&＃xff0f;频繁项集挖掘&＃xff1b;
main&＃xff0f;scala里面就是所有封装Spark源码中的代码&＃xff1b;
test&＃xff0f;scala里面对应每个封装代码的测试&＃xff1b;


2. 工程采用Maven构建&＃xff0c;直接根据pom文件加载对应依赖&＃xff1b;


3. 该工程需要经过maven打包&＃xff0c;把打包好的jar包放到CDH的虚拟机中的HDFS上某一固定目录&＃xff0c;方便Spark算法调用工程调用&＃xff08;具体目录下文有说&＃xff09;&＃xff1b;

3.1.2 单个算法实现&＃xff08;封装&＃xff0f;测试&＃xff09;&＃xff0c;比如针对逻辑回归

1. 针对逻辑回归&＃xff0c;其封装代码如下所示&＃xff1a;
代码清单3-1 逻辑回归算法封装&＃xff08;Scala&＃xff09;

package com.fz.classification
import com.fz.util.Utils
import org.apache.spark.mllib.classification.LogisticRegressionWithSGD, LogisticRegressionWithLBFGS
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.SparkConf, SparkContext
/**
* 逻辑回归封装算法
* Labels used in Logistic Regression should be 0, 1, ..., k - 1 for k classes multi-label classification problem
* 输入参数&＃xff1a;
* testOrNot : 是否是测试&＃xff0c;正常情况设置为false
* input&＃xff1a;输出数据&＃xff1b;
* minPartitions : 输入数据最小partition个数
* output&＃xff1a;输出路径
* targetIndex&＃xff1a;目标列所在下标&＃xff0c;从1开始
* splitter&＃xff1a;数据分隔符&＃xff1b;
* method&＃xff1a;使用逻辑回归算法&＃xff1a;"SGD" or "LBFGS"
* hasIntercept : 是否具有截距
* numClasses: 目标列类别个数&＃xff1b;
* Created by fanzhe on 2016/12/19.
*/
object LogisticRegression
def main (args: Array[String])
if(args.length !&＃61; 9)
println("Usage: com.fz.classification.LogisticRegression testOrNot input minPartitions output targetIndex " &＃43;
"splitter method hasIntercept numClasses")
System.exit(-1)

val testOrNot &＃61; args(0).toBoolean // 是否是测试&＃xff0c;sparkContext获取方式不一样, true 为test
val input &＃61; args(1)
val minPartitions &＃61; args(2).toInt
val output &＃61; args(3)
val targetIndex &＃61; args(4).toInt // 从1开始&＃xff0c;不是从0开始要注意
val splitter &＃61; args(5)
val method &＃61; args(6) //should be "SGD" or "LBFGS"
val hasIntercept &＃61; args(7).toBoolean
val numClasses &＃61; args(8).toInt
val sc &＃61; Utils.getSparkContext(testOrNot,"Logistic Create Model")
// construct data
// Load and parse the data
val training &＃61; Utils.getLabeledPointData(sc,input,minPartitions,splitter,targetIndex).cache()
// Run training algorithm to build the model
val model &＃61; method match
case "SGD" &＃61;> new LogisticRegressionWithSGD()
.setIntercept(hasIntercept)
.run(training)
case "LBFGS" &＃61;> new LogisticRegressionWithLBFGS().setNumClasses(numClasses)
.setIntercept(hasIntercept)
.run(training)
case _ &＃61;> throw new RuntimeException("no method")

// save model
model.save(sc,output)
sc.stop()

在上面的代码中&＃xff0c;有对每个参数的解释&＃xff0c;包括参数的含义&＃xff0c;参数有哪些参数等&＃xff1b;
在Main函数中&＃xff0c;首先对各个参数进行获取并赋值变量&＃xff0c;接着就是获取SparkContext&＃xff1b;
其中&＃xff0c;最重要的部分就是调用Spark自己封装的LogisticRegressionWithSGD 或 LogisticRegressionWithBFGS类进行逻辑回归建模&＃xff1b;
最后&＃xff0c;调用模型的save方法&＃xff0c;把模型固化到HDFS上&＃xff1b;
基本&＃xff0c;所有的算法封装都采取这种模式&＃xff0c;及对Spark MLlib中原生的算法再加一层封装。


2. 测试

测试主要使用JUnit进行测试&＃xff0c;其逻辑回归示例代码如下&＃xff1a;
代码清单3-2 逻辑回归算法封装测试&＃xff08;Scala&＃xff09;

package com.fz.classification
import java.io.File
import com.fz.util.Utils
import org.junit.Assert, Test
import Assert._
/**
* 测试Logistics Regression算法
* Created by fanzhe on 2016/12/19.
*/
&＃64;Test
class LogisticRegressionTest
&＃64;Test
def testMain1()&＃61;
// testOrNot input output targetIndex splitter method hasIntercept numClasses
val args &＃61; Array(
"true",
"./src/data/classification_regression/logistic.dat",
"2",
"./target/logistic/tmp1",
"1",
" ",
"SGD",
"true",
"2" // this parameter is useless
)
// 删除输出目录
Utils.deleteOutput(args(3))
LogisticRegression.main(args)
assertTrue(Utils.fileContainsClassName(args(3)&＃43;"/metadata/part-00000",
"org.apache.spark.mllib.classification.LogisticRegressionModel"))

&＃64;Test
def testMain2()&＃61;
// testOrNot input minPartitions output targetIndex splitter method hasIntercept numClasses
val args &＃61; Array(
"true",
"./src/data/classification_regression/logistic.dat",
"2",
"./target/logistic/tmp2",
"1",
" ",
"LBFGS",
"true",
"2"
)
// 删除输出目录
Utils.deleteOutput(args(3))
LogisticRegression.main(args)
assertTrue(Utils.fileContainsClassName(args(3)&＃43;"/metadata/part-00000",
"org.apache.spark.mllib.classification.LogisticRegressionModel"))

这里面的方法都是第一步先构建算法参数&＃xff1b;接着调用main方法&＃xff1b;第三步&＃xff0c;查看输出中是否具有模型的相关信息&＃xff1b;
当然&＃xff0c;这里面还可以添加多个测试方法&＃xff0c;使用不同的算法参数或数据进行测试&＃xff1b;&＃xff08;读者可自行添加&＃xff09;

3.2 Spark算法调用工程&＃xff1a;

3.2.1 界面介绍

1. 首页&＃xff1a;





在系统首页有对该系统实现算法的介绍&＃xff0c;系统主要功能有&＃xff1a;
1&＃xff09;集群参数维护&＃xff1a;主要是底层使用的Hadoop集群的参数配置&＃xff0c;每次配置完成后&＃xff0c;不仅仅会更新数据库对应记录&＃xff0c;而且会更新Hadoop Configuration的获取&＃xff1b;
2&＃xff09;监控&＃xff1a;主要指Spark任务运行在YARN资源管理器下的任务状态监控&＃xff1b;
3&＃xff09;文件上传及预览&＃xff1a;文件上传主要是上传本地测试数据到HDFS上&＃xff0c;方便页面进行测试&＃xff1b;而预览则是查看HDFS上面的数据&＃xff1b;
4&＃xff09;分类与回归&＃xff0f;协同过滤&＃xff0f;聚类&＃xff0f;降维&＃xff0f;关联规则&＃xff1a;各个种类算法下面的每个算法的调用建模页面&＃xff1b;
2. 集群参数页面&＃xff1a;



点击初始化&＃xff0c;会把各个参数固定写入到后台数据库中&＃xff0c;用户可以根据自己集群的配置不同&＃xff0c;而进行参数修改&＃xff0c;而每次修改也会刷新Hadoop 中Configuration的获取&＃xff1b;
3. 监控&＃xff1a;



监控页面&＃xff0c;会监控用户提交的SPark任务的运行状态&＃xff0c;如果任务失败&＃xff0c;则会显示异常信息&＃xff08;代码中只是截取了部分信息&＃xff0c;需要进行调整&＃xff0c;看如何可以得出重要的信息&＃xff0c;直接显示&＃xff09;&＃xff1b;后面会有具体实现过程分析。
4. 文件上传&＃xff1a;



文件上传有两个功能:1)可指定一个本地目录和一个HDFS目录&＃xff0c;然后把数据从本地上传到HDFS中&＃xff1b;2&＃xff09;直接选择对应算法的数据&＃xff0c;然后进行初始化&＃xff0c;这个是把本地工程路径src/main/data中的对应数据上传到HDFS中的固定目录中&＃xff1b;这两个上传的数据都可以在后面的算法建模中进行使用。
还有一点需要注意&＃xff1a;被写入的HDFS路径是需要具有写权限的&＃xff0c;而用户则是启动Tomcat的用户&＃xff1b;
5. 文件查看&＃xff1a;



文件查看功能只能查看Text编码的文件&＃xff0c;即文本文件&＃xff0c;同时可以输入行号&＃xff0c;即可进行文件内容的读取&＃xff1b;
6. 逻辑回归算法&＃xff1a;




在逻辑回归算法界面&＃xff0c;输入算法参数&＃xff0c;点击提交&＃xff0c;如果任务提交成功&＃xff0c;即可在下面看到任务提交的ID&＃xff0c;如果提交失败&＃xff08;即任务ID获取不到&＃xff09;&＃xff0c;同样有对应的提示信息&＃xff1b;
同时&＃xff0c;在任务提交后&＃xff0c;在监控界面同样可以观察到该任务的状态&＃xff0c;通过刷新即可获得最新的任务状态&＃xff1b;


7. 其他算法与逻辑回归算法类似

3.2.2 架构

系统架构图如下所示&＃xff08;算法调用及监控&＃xff09;&＃xff1a;



流程描述如下&＃xff1a;
1. 前台界面设置参数&＃xff0c;包括算法数据、算法参数等&＃xff0c;然后提交任务&＃xff1b;
2. 任务提交后&＃xff0c;CloudAction接收后&＃xff0c;会发起一个线程&＃xff0c;该线程会启动Hadoop上的一个Job&＃xff0c;该Job有一个返回值&＃xff0c;为任务ID&＃xff0c;如果任务提交失败&＃xff0c;则返回null&＃xff1b;
3. 初级监控状态&＃xff1a;CloudAction发起线程后&＃xff0c;主线程阻塞&＃xff0c;等待hadoop任务线程返回值&＃xff0c;根据返回值状态&＃xff0c;前台返回任务提交成功或失败&＃xff1b;
4. 在3的同时&＃xff0c;即可通过DBService来更新数据库相应表JobInfo的状态&＃xff1b;
5. 在monitor.html界面&＃xff0c;通过刷新按钮即可及时获取Hadoop任务状态&＃xff08;有相应的服务&＃xff0c;见下文介绍&＃xff09;&＃xff0c;并更新数据库相关数据&＃xff0c;返回前台所有任务信息&＃xff1b;

3.2.3 部分实现细节

1. Spark提交任务
参考《
基于Spark ALS在线推荐系统》&＃xff1b;


2. monitor实时查询任务状态列表
monitor实时查询任务状态列表其流程描述如下&＃xff1a;

1&＃xff09; 获取JobInfo中最新的records条记录&＃xff1b;
2&＃xff09; 查找其中isFinished字段为false的数据&＃xff1b;
3&＃xff09; 根据2&＃xff09;中查找的数据&＃xff0c;去YARN获取其实时状态&＃xff0c;并更新1&＃xff09;中的数据&＃xff0c;然后存入数据库中&＃xff1b;
4&＃xff09; 根据row和page字段分页返回JSON数据&＃xff1b;其代码如下所示&＃xff1a;
代码清单3-3 更新监控任务列表

public void getJobInfo()
Map jsonMap &＃61; new HashMap();
// 1.
List jobInfos &＃61; dBService.getLastNRows("JobInfo","jobId",true,records);
// 2,3
List list &＃61; null;
try
list &＃61; HUtils.updateJobInfo(jobInfos);
if(list !&＃61; null || list.size()>0)
dBService.updateTableData(list);

catch (Exception e)
e.printStackTrace();
log.warn("更新任务状态异常&＃xff01;");
jsonMap.put("total", 0);
jsonMap.put("rows", null);
Utils.write2PrintWriter(JSON.toJSONString(jsonMap));
return ;

// 4.
jsonMap.put("total",list.size());
jsonMap.put("rows",Utils.getSubList(list,page,rows));
Utils.write2PrintWriter(JSON.toJSONString(jsonMap));
第一步通过dBService获取给定records个记录&＃xff1b;第二步则更新这些记录&＃xff1b;看下HUtils.updateJobInfo的实现&＃xff1a;
代码清单3-4 获取任务最新状态

public static List updateJobInfo(List jobInfos)throws YarnException,IOException
List list &＃61; new ArrayList<>();
JobInfo jobInfo;
for(Object o :jobInfos)
jobInfo &＃61; (JobInfo) o;
if(!jobInfo.isFinished()) // 如果没有完成&＃xff0c;则检查其最新状态
ApplicationReport appReport&＃61;null;
try
appReport &＃61; getClient().getApplicationReport(SparkUtils.getAppId(jobInfo.getJobId()));
catch (YarnException | IOException e)
e.printStackTrace();
throw e;

/**
* NEW, 0
NEW_SAVING, 1
SUBMITTED, 2
ACCEPTED, 3
RUNNING, 4
FINISHED, 5
FAILED, 6
KILLED; 7
*/
switch (appReport.getYarnApplicationState().ordinal())
case 0 | 1 | 2 |3 : // 都更新为Accepted状态
jobInfo.setRunState(JobState.ACCETPED);
break;
case 4 :
jobInfo.setRunState(JobState.RUNNING);break;
case 5:
// UNDEFINED,
// SUCCEEDED,
// FAILED,
// KILLED;
switch (appReport.getFinalApplicationStatus().ordinal())
case 1: jobInfo.setRunState(JobState.SUCCESSED);
SparkUtils.cleanupStagingDir(jobInfo.getJobId());
jobInfo.setFinished(true);break;
case 2:
jobInfo.setRunState(JobState.FAILED);
SparkUtils.cleanupStagingDir(jobInfo.getJobId());
jobInfo.setErrorInfo(appReport.getDiagnostics().substring(0,Utils.EXCEPTIONMESSAGELENGTH));
jobInfo.setFinished(true);break;
case 3:
jobInfo.setRunState(JobState.KILLED);
SparkUtils.cleanupStagingDir(jobInfo.getJobId());
jobInfo.setFinished(true);break;
default: log.warn("App:" &＃43; jobInfo.getJobId() &＃43; "获取任务状态异常! " &＃43;
"appReport.getFinalApplicationStatus():"&＃43;appReport.getFinalApplicationStatus().name()
&＃43;",ordinal:"&＃43; appReport.getFinalApplicationStatus().ordinal());

break;
case 6:
jobInfo.setRunState(JobState.FAILED);
SparkUtils.cleanupStagingDir(jobInfo.getJobId());
jobInfo.setErrorInfo(appReport.getDiagnostics().substring(0,Utils.EXCEPTIONMESSAGELENGTH));
jobInfo.setFinished(true);break;
case 7:
jobInfo.setRunState(JobState.KILLED);
SparkUtils.cleanupStagingDir(jobInfo.getJobId());
jobInfo.setFinished(true);break;
default: log.warn("App:" &＃43; jobInfo.getJobId() &＃43; "获取任务状态异常!"&＃43;
"appReport.getYarnApplicationState():"&＃43;appReport.getYarnApplicationState().name()
&＃43;",ordinal:"&＃43; appReport.getYarnApplicationState().ordinal());

jobInfo.setModifiedTime(new Date());

list.add(jobInfo);// 把更新后的或原始的JobInfo添加到list中

return list;
这里的工作就是根据数据库中任务的状态&＃xff0c;只查询任务没有完成的任务的最新状态&＃xff0c;并更新原始任务状态&＃xff0c;最后把更新后的或者原始任务添加到list中&＃xff0c;并返回&＃xff1b;
在代码清单3-3中&＃xff0c;返回更新后的list后&＃xff0c;接着调用了DBService.updateTableData,对数据进行固化&＃xff1b;最后&＃xff0c;使用subList对list进行截取&＃xff0c;返回给前台某个分页的数据。

4. Spark算法调用工程后续开发&＃xff1a;

不得不说&＃xff0c;这个版本的工程还是没有开发完成的&＃xff0c;那如果你想接着来开发&＃xff0c;一般流程是怎样的呢&＃xff1f;
。。。

1&＃xff09;编写src/main/java/下算法对应的Thread&＃xff1b;
2&＃xff09;编写webapp下的对应页面&＃xff1b;
3&＃xff09;编写webapp/js下对应的js&＃xff1b;
4&＃xff09;修改webapp/preprocess/upload.jsp&＃xff0c;添加一条数据上传记录&＃xff0c;并在main/data下添加对应的数据&＃xff1b;
5&＃xff09;启动工程&＃xff0c;在页面上传数据&＃xff0c;然后选择算法&＃xff0c;设置参数&＃xff0c;即可提交任务&＃xff0c;提交任务后在监控界面即可看到算法运行状态&＃xff1b;



工程状态&＃xff08;假设Scala工程为工程1&＃xff0c;调用Spark算法工程为工程2&＃xff09;&＃xff1a;
工程1&＃xff1a;
基本封装了Spark Mllib中的数据挖掘相关算法&＃xff0c;包括聚类、分类、回归、协同过滤、降维、频繁集挖掘&＃xff08;这个还有点问题&＃xff09;&＃xff1b;
工程2&＃xff1a;
目前只做了分类和回归算法的相关页面以及调用&＃xff1b;

所以&＃xff0c;如果你要在这个版本上开发&＃xff0c;那么可以参考上面的流程先试着编写ALS算法的调用即可。

5. 总结

1. Spark算法调用工程还有很多页面没有完成&＃xff0c;这个是类似重复性工作&＃xff0c;并没有难点需要克服&＃xff1b;
2. Spark算法调用工程中针对每个算法&＃xff0c;本来是想在其算法调用界面加上其数据描述、算法描述、参数描述的&＃xff0c;不过暂时还没有添加&＃xff0c;but这些信息在Scala算法封装工程里面都有&＃xff1b;
3. 关于使用SPARK ON YARN的方式调用Spark算法&＃xff0c;并使用YARN来管理任务的流程基本在Spark算法调用工程中体现淋漓尽致了&＃xff0c;再多也玩不出花儿了&＃xff0c;所以如果有想学习研究这块内容的&＃xff0c;则工程是一个很好的参考&＃xff1b;
4. 之前对于分类算法这块是想加算法对比分析的&＃xff0c;然后再加上些图表之类的展示&＃xff0c;这样就显得更加高大上了&＃xff0c;不过目前只进行了一步&＃xff0c;就是写了个分类算法评估的Scala封装算法&＃xff1b;
5. 可以考虑一些流程性的定时任务之类的加入到工程中&＃xff0c;这样其实有点像Oozie了&＃xff0c;不过为什么Oozie里面没有直接拖拽界面或流程任务监控管理的东西&＃xff0c;如果有的话其实就更加像一个商业的软件了&＃xff08;Kettle&＃xff09;&＃xff1b;
6. 关于SSH框架其实我是比较弱的&＃xff0c;所以里面应用ssh的地方只是简单的应用&＃xff08;比如说在返回分页的时候&＃xff0c;我直接用的是subList&＃xff0c;这个应该是不妥的&＃xff09;&＃xff1b;
7. 关于前台页面展现&＃xff0c;我也是比较弱的&＃xff0c;所以界面风格或单页的相关信息显示之类的&＃xff0c;看着还不能做到赏心悦目&＃xff1b;
8. The Code is free &＃xff0c;just enjoy&＃xff01;






分享&＃xff0c;成长&＃xff0c;快乐





脚踏实地&＃xff0c;专注





转载请注明blog地址&＃xff1a;http://blog.csdn.net/fansy1990